package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

public class Test2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);
        DataStream<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("dwd_page_yk7", "test"));
        //last_page_id 必为 null。过滤 last_page_id != null 的数据，减小数据量，提升计算效率
        SingleOutputStreamOperator<JSONObject> etlStream = stream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                System.out.println("s = " + s);
                JSONObject jsonObject = JSON.parseObject(s);
                JSONObject page = jsonObject.getJSONObject("page");
                String last_page_id = page.getString("last_page_id");
                if (last_page_id == null) {
                    collector.collect(jsonObject);
                }
            }
        });

        SingleOutputStreamOperator<JSONObject> watermarksStream = etlStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts");
            }
        }));

        /*
        5）流量域用户跳出主题，flink读取流量域页面浏览主题用CEP实现跳出：跳出是指会话中只有一个页面的访问行为，如果能获取会话的所有页面，只要筛选页面数为 1 的会话即可获取跳出明细数据（5分）


         */
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                return lastPageId == null || lastPageId.length() <= 0;
            }
        }).within(org.apache.flink.streaming.api.windowing.time.Time.seconds(10));

        // 把CEP作用到流上,并进行分组
        PatternStream<JSONObject> cepDS = CEP.pattern(watermarksStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                String mid = jsonObject.getJSONObject("common").getString("mid");
                return mid;
            }
        }), pattern);

        // 提取两部分数据
        OutputTag<JSONObject> timeOutTag = new OutputTag<JSONObject>("timeout") {};

        SingleOutputStreamOperator<JSONObject> selectDS = cepDS.select(timeOutTag, new PatternTimeoutFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                return map.get("start").get(0);
            }
        }, new PatternSelectFunction<JSONObject, JSONObject>() {
            @Override
            public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                return map.get("start").get(0);
            }
        });

        // 超时数据
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(timeOutTag);

        // union 把超时数据和满足CEP规则数据合并
        DataStream<JSONObject> unionDS = selectDS.union(timeOutDS);

        unionDS.print();

        env.execute();


    }
}
